package com.wyh.apitest.sink

import com.wyh.apitest.source.SensorReading
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.redis.RedisSink
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}

object RedisSinkDemo {
  def main(args: Array[String]): Unit = {
    //定义流式处理环境
    val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(1)

    val sensorinfo: DataStream[String] = streamEnv.readTextFile("D:\\IdeaProjects\\FlinkStudyScala\\src\\main\\resources\\sensor.txt")
    //转换成样例类类型
    val dataStream = sensorinfo.map(s => {
      val arr: Array[String] = s.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })

    //定义一个FlinkJedisPoolConfig
    val conf = new FlinkJedisPoolConfig.Builder()
        .setHost("192.168.205.100")
        .setPort(6379)
        .build()

    dataStream.addSink(new RedisSink[SensorReading](conf,new MyRedisMapper))
    streamEnv.execute("Redis Sink Demo")
  }

}


//定义一个RedisMapper
class MyRedisMapper extends RedisMapper[SensorReading]{

  //定义数据写入的Redis命令  HSET 表名 key value
  override def getCommandDescription: RedisCommandDescription = {
    new RedisCommandDescription(RedisCommand.HSET,"sensor_temp")
  }

  //指定key 将id指定为key
  override def getKeyFromData(t: SensorReading): String = {
    t.id
  }


  //指定value 将温度值指定为value
  override def getValueFromData(t: SensorReading): String = {
    t.temperature.toString
  }
}
